package com.lili.collection.jdbc.handler;

import com.lili.collection.core.domain.CollectCommonConfig;
import com.lili.collection.core.fields.Field;
import com.lili.collection.core.record.Record;
import com.lili.collection.core.runner.result.RecordInfo;
import com.lili.collection.core.utils.NoticeUtil;
import com.lili.collection.jdbc.utils.JdbcCharUtil;
import com.lili.collection.jdbc.utils.JdbcUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;

/**
 *描述: 用于JDBC合并的处理器
 *@date: 2024/1/4  11:11
 *@author: lili
 */
@Slf4j
public class JDBCMergeIntoHandler {
    private String tableName;
    private CollectCommonConfig config;
    private Connection connection;
    private static final String SUB_WHERE_SQL="SUB_WHERE_SQL";
    private static final String KEY_LIST_SQL="KEY_LIST_SQL";
    private static final String MYSQL="mysql";
    private boolean isIncrementalMode;
    private boolean isMySQL;
    private Set<String> primaryKeySet;
    private RecordInfo recordInfo;

    public JDBCMergeIntoHandler(Connection connection,CollectCommonConfig config){
        this.connection=connection;
        this.config=config;
        this.isIncrementalMode=this.config.isIncrementalMode();
        this.isMySQL=isMysqlDb();
        this.primaryKeySet = this.config.getPrimaryKeySet();
    }

    public void setTableName(String tableName) {
        this.tableName = tableName;
    }

    public void setRecordInfo(RecordInfo recordInfo) {
        this.recordInfo = recordInfo;
    }

    private boolean isMysqlDb(){
        try {
            return MYSQL.equalsIgnoreCase(connection.getMetaData().getDatabaseProductName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 采用批量机制进行更新已存在的数据
     */
    public void mergeIntoWithJdbc(List<Record> output) {
        if(!isIncrementalMode || CollectionUtils.isEmpty(output) || CollectionUtils.isEmpty(primaryKeySet)){
            return;
        }
        //只有自增模式才使用合并功能
        List<List<Record>> partition = ListUtils.partition(output, 500);
        for (int k = 0; k < partition.size(); k++) {
            List<Record> records = partition.get(k);
            Map<String, List<Object>> parameterMap = collectPrimaryKeyValue(records);
            Map<String,Object> dataMap=new HashMap<>(2);
            if(this.primaryKeyCheck(tableName, parameterMap, dataMap)==0){
                continue;
            }
            Set<String> primarySet = getPrimaryKeySet(parameterMap, dataMap);
            if(CollectionUtils.isEmpty(primarySet)){
                continue;
            }
            PreparedStatement updateStatement = null;
            List<Record> updateRecordList=new ArrayList<>();
            try {
                String updatePrepareSQL = createUpdatePrepareSQL(records.get(0).getField().getValueAsMap(),primaryKeySet,tableName);
                updateStatement = connection.prepareStatement(updatePrepareSQL);
                Iterator<Record> iterator = records.iterator();
                while (iterator.hasNext()) {
                    Record record = iterator.next();
                    Map<String, Field> valueAsMap = record.getField().getValueAsMap();
                    String primaryKey = getPrimaryKey(valueAsMap);
                    if(primarySet.contains(primaryKey)){
                        List<Object> whereList=new ArrayList<>();
                        List<Object> updateList=new ArrayList<>();
                        createUpdateValue(valueAsMap, primaryKeySet, whereList, updateList,tableName);
                        //创建修改语句
                        updateList.addAll(whereList);
                        for (int i = 1; i <= updateList.size(); i++) {
                            updateStatement.setObject(i, JdbcCharUtil.handleEmptyChar(updateList.get(i-1)));
                        }
                        updateStatement.addBatch();
                        this.recordInfo.incrementSuccessCount();
                        updateRecordList.add(record);
                        iterator.remove();
                    }
                }
                updateStatement.executeBatch();
                updateStatement.clearBatch();
                NoticeUtil.saveNotice(config,updateRecordList,"Y");
            } catch (SQLException e) {
                NoticeUtil.saveNotice(config,updateRecordList,"N");
                throw new RuntimeException(e);
            } finally {
                JdbcUtil.closeDb(null, updateStatement, null);
            }
        }
    }

    /**
     * 在单行数据中获取主键
     * @param valueAsMap
     * @return 主键
     */
    private String getPrimaryKey(Map<String, Field> valueAsMap) {
        StringBuilder primaryKeyBuilder=new StringBuilder();
        for (String key : primaryKeySet) {
            primaryKeyBuilder.append(valueAsMap.get(key).getValue());
        }
        return primaryKeyBuilder.toString();
    }

    /**
     * 在截断的行数据中收集主键值（兼容联合主键）
     * @param records 行数
     * @return 收集主键值
     */
    private Map<String, List<Object>> collectPrimaryKeyValue(List<Record> records) {
        Map<String,List<Object>> parameterMap=new LinkedHashMap<>();
        for (Record record : records) {
            Map<String, Field> valueAsMap = record.getField().getValueAsMap();
            for (String primaryKey : primaryKeySet) {
                Field field = valueAsMap.get(primaryKey);
                if(field==null || field.getValue()==null){
                    throw new IllegalArgumentException("当前主键【"+primaryKey+"】值为空，请核实！");
                }
                List<Object> objects = parameterMap.get(primaryKey);
                if(CollectionUtils.isNotEmpty(objects)){
                    objects.add(field.getValue());
                } else {
                    objects=new ArrayList<>();
                    objects.add(field.getValue());
                    parameterMap.put(primaryKey,objects);
                }
            }
        }
        return parameterMap;
    }

    /**
     * 创建预编译更新语句
     * @param valueMap 单行值
     * @param primarySet 主键值
     * @param tableName 表名
     * @return
     */
    private String createUpdatePrepareSQL(Map<String, Field> valueMap,Set<String> primarySet,String tableName){
        StringBuilder updateBuilder=new StringBuilder();
        StringBuilder whereBuilder=new StringBuilder();
        int updateIndex=0;
        int whereIndex=0;
        for (Map.Entry<String, Field> entry : valueMap.entrySet()) {
            String key = entry.getKey();
            if(primarySet.contains(key)){
                //主键
                if(whereIndex!=0){
                    whereBuilder.append(" and ");
                }
                if(isMySQL){
                    key="`"+key+"`";
                }
                whereBuilder.append(key).append("=").append("?");
                whereIndex++;
            }else {
                //修改key-value
                if(updateIndex!=0){
                    updateBuilder.append(" , ");
                }
                if(isMySQL){
                    key="`"+key+"`";
                }
                updateBuilder.append(key).append("=").append("?");
                updateIndex++;
            }
        }
        if(updateIndex==0){
            throw new IllegalArgumentException("更新语句不能为空！");
        }
        if(whereIndex==0){
            throw new IllegalArgumentException("更新条件不能为空！");
        }
        StringBuilder partUpdateSqlBuilder=new StringBuilder();
        partUpdateSqlBuilder.append(updateBuilder);
        partUpdateSqlBuilder.append(" where ");
        partUpdateSqlBuilder.append(whereBuilder);

        StringBuilder updateSQLBuilder=new StringBuilder();
        updateSQLBuilder.append("update ")
                .append(tableName)
                .append(" set ")
                .append(partUpdateSqlBuilder);
        return updateSQLBuilder.toString();
    }

    private void createUpdateValue(Map<String, Field> valueMap,Set<String> primarySet,List<Object> whereList,List<Object> updateList,String tableName){
        int updateIndex=0;
        int whereIndex=0;
        for (Map.Entry<String, Field> entry : valueMap.entrySet()) {
            String key = entry.getKey();
            if(primarySet.contains(key)){
                //主键
                whereList.add(entry.getValue().getValue());
                whereIndex++;
            }else {
                //修改key-value
                updateList.add(entry.getValue().getValue());
                updateIndex++;
            }
        }
        if(updateIndex==0){
            throw new IllegalArgumentException("更新语句不能为空！");
        }
        if(whereIndex==0){
            throw new IllegalArgumentException("更新条件不能为空！");
        }
    }

    private Set<String> getPrimaryKeySet(Map<String, List<Object>> parameterMap, Map<String, Object> dataMap){
        String subWhereSql = (String) dataMap.get(SUB_WHERE_SQL);
        String keyListSql = (String) dataMap.get(KEY_LIST_SQL);
        if(StringUtils.isBlank(subWhereSql) || StringUtils.isBlank(keyListSql)){
            throw new IllegalArgumentException("条件或者key为空！");
        }
        Set<String> primarySet = new HashSet<>();
        ResultSet resultSet=null;
        PreparedStatement batchSelectState=null;
        try {
            StringBuilder selectBuilderSql=new StringBuilder();
            selectBuilderSql.append("select ")
                    .append(keyListSql)
                    .append(" from ")
                    .append(tableName)
                    .append(" where ")
                    .append(subWhereSql);
            String selectSQL = selectBuilderSql.toString();
            batchSelectState = connection.prepareStatement(selectSQL);
            int paramIndex = 0;
            for (List<Object> valueList : parameterMap.values()) {
                for (Object value : valueList) {
                    batchSelectState.setObject(++paramIndex, JdbcCharUtil.handleEmptyChar(value));
                }
            }
            resultSet = batchSelectState.executeQuery();
            while(resultSet.next()){
                StringBuilder jointPrimaryKey=new StringBuilder();
                for (String column : parameterMap.keySet()) {
                    Object object = resultSet.getObject(column);
                    if(object==null){
                        throw new IllegalArgumentException("主键【"+column+"】不能为空！");
                    }
                    jointPrimaryKey.append(object);
                }
                primarySet.add(jointPrimaryKey.toString());
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            JdbcUtil.closeDb(resultSet, batchSelectState, null);
        }
        return primarySet;
    }

    private int primaryKeyCheck(String tableName, Map<String, List<Object>> parameterMap,Map<String,Object> dataMap) {
        String batchCountSql = this.createBatchCountSql(parameterMap, tableName,dataMap);
        log.info("批量验证SQl为:{}",batchCountSql);
        PreparedStatement batchCountState=null;
        ResultSet resultSet = null;
        try {
            batchCountState = connection.prepareStatement(batchCountSql);
            int paramIndex = 0;
            for (List<Object> valueList : parameterMap.values()) {
                for (Object value : valueList) {
                    batchCountState.setObject(++paramIndex, JdbcCharUtil.handleEmptyChar(value));
                }
            }
            resultSet=batchCountState.executeQuery();
            if(resultSet.next()){
                int realTotal = resultSet.getInt("real_total");
                int total = resultSet.getInt("total");
                if(realTotal!=total){
                    throw new IllegalArgumentException("表【"+tableName+"】的字段【"+ parameterMap.keySet()+"】非主键字段，请核实！");
                }
                return total;
            }
        }catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            JdbcUtil.closeDb(resultSet, batchCountState, null);
        }
        return 0;
    }

    private String createBatchCountSql(Map<String, List<Object>> parameterMap,String tableName,Map<String,Object> dataMap) {
        StringBuilder sqlBuilder=new StringBuilder("select ");
        String whereConditionSQL = createWhereCondition(parameterMap);
        String keyListStr = getKeyListStr(parameterMap);
        sqlBuilder.append("count(distinct").append(" ").append(keyListStr)
                .append(") as real_total,count(*) as total from ")
                .append(tableName)
                .append(" where ")
                .append(whereConditionSQL);
        dataMap.put(SUB_WHERE_SQL,whereConditionSQL);
        dataMap.put(KEY_LIST_SQL,keyListStr);
        return sqlBuilder.toString();
    }

    private String createWhereCondition(Map<String, List<Object>> parameterMap){
        List<String> subConditionList=new ArrayList<>();
        for (Map.Entry<String, List<Object>> entry : parameterMap.entrySet()) {
            String column = entry.getKey();
            if(isMySQL){
                column="`"+column+"`";
            }
            List<Object> valueList = entry.getValue();
            if(CollectionUtils.isEmpty(valueList)){
                throw new IllegalArgumentException("当前列【"+column+"】无值！");
            }
            StringBuilder whereBuilder=new StringBuilder(column);
            whereBuilder.append(" ")
                    .append("in(")
                    .append(getValueListStr(valueList))
                    .append(")");
            subConditionList.add(whereBuilder.toString());
        }
        if(CollectionUtils.isEmpty(subConditionList)){
            throw new IllegalArgumentException("where的子条件为空！");
        }
        StringBuilder whereBuilder=new StringBuilder();
        for (int i = 0; i < subConditionList.size(); i++) {
            String subCondition=subConditionList.get(i);
            if(i!=0){
                whereBuilder.append(" and ");
            }
            whereBuilder.append(subCondition);
        }
        return whereBuilder.toString();
    }

    private String getKeyListStr(Map<String, List<Object>> parameterMap){
        StringBuilder columnBuilder=new StringBuilder();
        for (String column : parameterMap.keySet()) {
            if(isMySQL){
                column="`"+column+"`";
            }
            columnBuilder.append(column).append(",");
        }
        StringBuilder builder = columnBuilder.deleteCharAt(columnBuilder.length() - 1);
        return builder.toString();
    }
    private String getValueListStr(List<Object> valueList){
        StringBuilder valueBuilder=new StringBuilder();
        for (int i = 0; i < valueList.size(); i++) {
            valueBuilder.append("?").append(",");
        }
        StringBuilder builder = valueBuilder.deleteCharAt(valueBuilder.length() - 1);
        return builder.toString();
    }
}